package org.solrcn.hadoop.index;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.lucene.analysis.cjk.CJKAnalyzer;
import org.apache.lucene.codecs.PostingsFormat;
import org.apache.lucene.codecs.bloom.BloomFilteringPostingsFormat;
import org.apache.lucene.codecs.bloom.DefaultBloomFilterFactory;
import org.apache.lucene.codecs.lucene41.Lucene41PostingsFormat;
import org.apache.lucene.codecs.lucene42.Lucene42Codec;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.document.IntField;
import org.apache.lucene.document.LongField;
import org.apache.lucene.document.NumericDocValuesField;
import org.apache.lucene.document.StringField;
import org.apache.lucene.document.TextField;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.IndexWriterConfig;
import org.apache.lucene.index.NoMergePolicy;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.NoLockFactory;
import org.apache.lucene.util.Version;
import org.joda.time.DateTime;
import org.solrcn.lucene.hdfs.HdfsDirectory;

import com.google.common.hash.Hashing;



public class LuceneMR extends Configured implements Tool {
	final static Version matchVersion = Version.LUCENE_43;

	//获得写HDFS的Writer
	private static IndexWriter getHDFSWriter(final String outputDir) throws IOException {
		HdfsDirectory directory = new HdfsDirectory(outputDir);
		directory.setLockFactory(NoLockFactory.getNoLockFactory());
		IndexWriterConfig iwc = new IndexWriterConfig(matchVersion, new CJKAnalyzer(matchVersion));
		iwc.setMergePolicy(NoMergePolicy.NO_COMPOUND_FILES);
		iwc.setRAMBufferSizeMB(512);
		iwc.setMaxBufferedDocs(Integer.MAX_VALUE);
		
		
		final HashMap<String,PostingsFormat> postingsFormatForField = new HashMap<String,PostingsFormat>();
		Lucene42Codec lucene42Codec = new Lucene42Codec() {
			BloomFilteringPostingsFormat postingOptions = new BloomFilteringPostingsFormat(new Lucene41PostingsFormat(), new DefaultBloomFilterFactory());
			
			@Override
			public PostingsFormat getPostingsFormatForField(String field) {
				if(postingsFormatForField.containsKey(field)){
					return postingsFormatForField.get(field);
				}
				return postingOptions;
			}
		};		
		
		iwc.setCodec(lucene42Codec);
		return new IndexWriter(directory, iwc);
	}

	//solr分布式 _version_ 字段值
	private static long getVersion() {
		long time = System.currentTimeMillis();
		return time << 20;
	}

	public static class MapClass extends Mapper<LongWritable, Text, Text, Text> {

		OutputCollector<Text, Text> oc = null;
		ArrayList<Document> docs = new ArrayList<Document>();
		String outputDir;
		IndexWriter writer;
		String fileName;
		DateTime dt;
		Random r = new Random();
		int collectionNO;
		
		@Override
		public void setup(Context context) throws IOException {
			
			String indexId = "";
			collectionNO = context.getConfiguration().getInt("collectionNO", 1);
			indexId = Hashing.murmur3_32().hashLong((System.currentTimeMillis()+r.nextInt(1000))).toString();
			
			outputDir = context.getConfiguration().get("mapred.output.dir") + "/index_" + indexId;
			
			InputSplit inputSplit = context.getInputSplit();
			fileName = ((FileSplit) inputSplit).getPath().toString();

			writer = getHDFSWriter(outputDir);
			int year = (collectionNO / 24 / 30 / 12) % 10 + 2010;
			int month = ((collectionNO / 24) / 30) % 12 + 1;
			int day = (collectionNO / 24) % 30 + 1;
			int hour = collectionNO % 24;
			
			dt = new DateTime(year, month, day, hour, 0, 0);
		}

		@Override
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

			Document doc = new Document();

//			StringField idField = new StringField("id", fileName.hashCode() + key.toString(), Field.Store.YES);
			//用UUID作为SOLR的唯一标识字段值
			StringField idField = new StringField("id", java.util.UUID.randomUUID().toString(), Field.Store.YES);

			doc.add(idField);
			String[] str = value.toString().split("\t");

			for (int i = 0; i < str.length; i++) {
				if (str[i].trim().isEmpty())
					continue;
				doc.add(new TextField("data_" + i + "_cjk", str[i], Field.Store.NO));
			}
			
			//随机产生一小时内任意秒数
			long rndTime = r.nextInt(3600)*1000;
			//普通存储long和int
			doc.add(new LongField("time_store_long", dt.plus(rndTime).getMillis(), Field.Store.YES));						
			doc.add(new IntField("time_store_int", dt.plus(rndTime).getMillisOfSecond(), Field.Store.YES));
			//DocValue属性long和int
			doc.add(new NumericDocValuesField("time_docvalue_int", dt.plus(rndTime).getMillis()));
			doc.add(new NumericDocValuesField("time_docvalue_long", dt.plus(rndTime).getMillisOfSecond()));
			//不存储long和int
			doc.add(new LongField("time_nostore_long", dt.plus(rndTime).getMillis(), Field.Store.NO));
			doc.add(new IntField("time_nostore_int", dt.plus(rndTime).getMillisOfSecond(), Field.Store.NO));
			
			doc.add(new LongField("_version_", getVersion(), Field.Store.YES));

			docs.add(doc);

			if (docs.size() > 10000) {
				writer.addDocuments(docs);
				docs.clear();
				docs = new ArrayList<Document>();
			}

		}

		@Override
		public void cleanup(Context context) throws IOException {

			if (docs.size() > 0) {
				writer.addDocuments(docs);
				docs.clear();
			}

			writer.commit();
			writer.close();
			try {
				context.write(new Text("Index"), new Text(outputDir));
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

		// public void close() throws IOException {
		// oc.collect(new Text("Index"), new Text(outputDir));
		// }
	}

	public static class Reduce extends Reducer<Text, Text, Text, IntWritable> {
		public int max = -1;
		String outputDir;
		Version matchVersion = Version.LUCENE_43;
		IndexWriter writer;

		@Override
		public void setup(Context context) throws IOException {
			outputDir = context.getConfiguration().get("mapred.output.dir") + "/index";
			writer = getHDFSWriter(outputDir);
		}

		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
			List<HdfsDirectory> dirs = new ArrayList<HdfsDirectory>();
			for (Text value : values) {
				dirs.add(new HdfsDirectory(value.toString()));
			}
			Directory[] dir = new Directory[dirs.size()];
			for (int i = 0; i < dir.length; i++) {
				dir[i] = dirs.get(i);
			}
			writer.addIndexes(dir);
			writer.forceMerge(1);
			writer.close();
		}

		@Override
		public void cleanup(Context context) throws IOException {
			if (writer != null)
				writer.close();
		}

	}

	public static class AgePartitioner extends Partitioner<Text, Text> {
		@Override
		public int getPartition(Text key, Text value, int numReduceTasks) {
			return 0;
//			String[] str = value.toString().split("\t");
//			int age = Integer.parseInt(str[1]);
//			if (numReduceTasks == 0) {
//				return 0;
//			}
//			if (age <= 20)
//				return 0;
//			else if (age > 20 && age <= 50)
//				return 1 % numReduceTasks;
//			else
//				return 2 % numReduceTasks;
		}
	}

	@Override
	public void setConf(Configuration conf) {

	}

	@Override
	public Configuration getConf() {
		return new Configuration();
	}

	@Override
	public int run(String[] args) throws Exception {
		Configuration conf = getConf();
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if (otherArgs.length < 2) {
			System.err.println("Usage: LuceneMR <in> <out> [collectionNO=1]");
			System.exit(2);
		}
		
		for (int i = 2; i < otherArgs.length; i++) {
			conf.set(otherArgs[i].split("=")[0], otherArgs[i].split("=")[1]);
		}
		
		Job job = new Job(conf, "Lucene Index v3.1");
		job.setJarByClass(LuceneMR.class);
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileInputFormat.setMinInputSplitSize(job, 512 * 1024 * 1024L);
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		job.setMapperClass(MapClass.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		// job.setCombinerClass(Reduce.class);
		job.setPartitionerClass(AgePartitioner.class);
		job.setReducerClass(Reduce.class);
		job.setNumReduceTasks(0);
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(NullOutputFormat.class);
		// job.setOutputFormatClass(TextOutputFormat.class);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(NullWritable.class);

		System.exit(job.waitForCompletion(true) ? 0 : 1);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new LuceneMR(), args);
		System.exit(res);
	}

}
